package golangAps

import (
	"context"
	"errors"
	"github.com/nutsdb/nutsdb"
	"os"
	"strings"
	"sync"
)

// 关闭

type Scheduler struct {
	// Job store
	store *StoreNutsdb
	// It should not be set manually.
	isRunning bool
	cancel    context.CancelFunc
	storeJobs sync.Map // map[string]*StoreJob
	mutexS    sync.RWMutex
}

// NewScheduler 默认创建一个 MemoryStore
func NewScheduler() *Scheduler {

	return &Scheduler{
		store:     NewStore(""),
		mutexS:    sync.RWMutex{},
		storeJobs: sync.Map{},
	}
}

func (s *Scheduler) IsRunning() bool {
	//s.mutexS.RLock()
	//defer s.mutexS.RUnlock()

	return s.isRunning
}

// Bind the store
func (s *Scheduler) SetStore(bucket string) (err error) {

	s.mutexS.Lock()
	defer s.mutexS.Unlock()
	if _, ok := s.storeJobs.Load(bucket); ok {
		return nil
	}
	// 不存在就创建， 存在就不管了
	storeJobs, err := NewStoreJob(s.store, bucket)
	if err != nil {
		if errors.Is(err, nutsdb.ErrBucketAlreadyExist) {
			return nil
		}
		return err
	}

	defer func() {
		EventChan <- EventInfo{
			EventCode: EVENT_JOBSTORE_ADDED,
			Error:     err,
			Msg:       strings.Join([]string{"store name: ", bucket}, ""),
		}
	}()
	storeJobs.Start()
	s.storeJobs.Store(bucket, storeJobs)

	return
}

// RemoveStore remove store
func (s *Scheduler) RemoveStore(bucket string) (err error) {
	defer func() {
		EventChan <- EventInfo{
			EventCode: EVENT_JOBSTORE_REMOVED,
			Error:     err,
			Msg:       strings.Join([]string{"store name: ", bucket}, ""),
		}
	}()

	if sj, ok := s.storeJobs.LoadAndDelete(bucket); ok {
		sj.(*StoreJob).Stop()
	}
	// bucket 都删除了， 就不用管任务是不是在运行了
	return s.store.Clear(bucket)
}

func (s *Scheduler) ExistStore(bucket string) (exist bool) {
	s.store.db.View(func(tx *nutsdb.Tx) error {
		exist = tx.ExistBucket(DefaultDss[0], bucket)
		return nil
	})
	return
}

// GetAllStoreName 获取当前所有的 store name
func (s *Scheduler) GetAllStoreName() ([]string, error) {
	buckets, err := s.store.GetAllBuckets()
	if err != nil {
		DefaultLog.Error(context.Background(), "Get all buckets failed", "error", err.Error())
		return nil, err
	}
	return buckets, nil
}

// Start scheduler 开启运行
func (s *Scheduler) Start() {
	if s.isRunning {
		DefaultLog.Info(context.Background(), "Scheduler is running.")
		return
	}

	s.isRunning = true
	ctx, cancel := context.WithCancel(context.Background())
	s.cancel = cancel
	StartEventsListen(ctx)
	buckets, err := s.GetAllStoreName()
	if err != nil {
		os.Exit(1)
		return
	}

	for _, bucket := range buckets {
		if _, ok := s.storeJobs.Load(bucket); ok {
			// 存在了就不要在启动了
			continue
		}
		sj, err := NewStoreJob(s.store, bucket)
		if err != nil {
			DefaultLog.Error(context.Background(), "Create store job failed", "error", err.Error())
			continue
		}
		sj.Start()
		s.storeJobs.Store(bucket, sj)
	}

	DefaultLog.Info(context.Background(), "Scheduler start.")
}

// Stop 停止scheduler
func (s *Scheduler) Stop() {
	if !s.isRunning {
		DefaultLog.Info(context.Background(), "Scheduler has stopped.")
		return
	}
	s.storeJobs.Range(func(k, v interface{}) bool {
		v.(*StoreJob).Stop()
		return true
	})
	err := s.store.Merge()
	if err != nil {
		DefaultLog.Error(context.Background(), "Scheduler merge failed", "error", err.Error())
	}
	err = s.store.Close()
	if err != nil {
		DefaultLog.Error(context.Background(), "Close store failed", "error", err.Error())
	}
	s.cancel()
	s.isRunning = false
	DefaultLog.Info(context.Background(), "Scheduler stop.")
}
